package com.zetyun.tiger.connector.op.v3;


import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Sets;
import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.sdk.action.MapAction;
import com.zetyun.rt.sdk.model.RtEvent;
import com.zetyun.rt.sdk.operator.OperatorContext;
import com.zetyun.tiger.connector.cdc.SimpleRedisContainer;
import lombok.SneakyThrows;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;

import java.util.Set;

import static com.zetyun.tiger.connector.op.v2.RedisOpV2Action.build;

@ActionMeta(
        id = "tiger.redis-opv3",
        tags = {"redis"},
        category = "map",
        name = "RedisOpV3",
        description = "统计客户经理、公共揽存、分支机构今日客户数",
        parameterClass = RedisOpV3Parameter.class,
        status = ActionMeta.Status.PUBLIC
)
public class RedisOpV3Action extends MapAction {

    // 公共揽存客户关系缓存
    private SimpleRedisContainer pbRelationshipDb;

    // 客户经理客户关系缓存
    private SimpleRedisContainer conntrRelationshipDb;

    // 实时数据库
    private SimpleRedisContainer rtDb;

    // 公共揽存客户关系缓存前缀
    private String pbRelationshipPrefix;

    // 公共揽存今日客户数
    private String pbCntPrefix;

    // 客户经理客户关系缓存前缀
    private String conntrRelationshipPrefix;

    // 客户经理今日客户数
    private String conntrCntPrefix;

    // 机构今日客户数
//    private String orgCntPrefix;

    // 叶子机构今日开户客户号缓存
    private String orgOpacCustPrefix;

    // 叶子机构今日销户客户号缓存
    private String orgCacctCustPrefix;

    private RedisOpV3Parameter param;

    private ObjectMapper om;

    private final Integer ttl = 60 * 60 * 24;  // 一天过期时间，仅 rtDb 需要设置


    @Override
    public void init(OperatorContext context) {
        super.init(context);
        om = new ObjectMapper();
        param = context.getParameters();
        pbRelationshipDb = buildContainer(param.getPbRelationshipDb());
        conntrRelationshipDb = buildContainer(param.getConntrRelationshipDb());
        rtDb = buildContainer(param.getRtDb());
        pbRelationshipPrefix = param.getPbRelationshipPrefix();
        pbCntPrefix = param.getPbCntPrefix();
        conntrRelationshipPrefix = param.getConntrRelationshipPrefix();
        conntrCntPrefix = param.getConntrCntPrefix();
//        orgCntPrefix = param.getOrgCntPrefix();
        orgOpacCustPrefix = param.getOrgOpacCustPrefix();
        orgCacctCustPrefix = param.getOrgCacctCustPrefix();
    }

    @Override
    public RtEvent apply(RtEvent in) throws Exception {

        String dayId = in.getField(param.getDayId(), String.class);
        String opacInstnNo = in.getField(param.getOpacInstnNo(), String.class);
        String paOpunCod = in.getField(param.getPaOpunCod(), String.class);
        String conntrNo = in.getField(param.getConntrNo(), String.class);
        String custNo = in.getField(param.getCustNo(), String.class);
        String acctNo = in.getField(param.getAcctNo(), String.class);
        String opacType = in.getField(param.getOpacType(), String.class);

        // 这笔业务归属的真实机构
        String realOrgCode = paOpunCod.isEmpty() ? opacInstnNo : paOpunCod;

        // 客户经理客户关系缓存
        String conntrRelationshipCacheName = conntrRelationshipPrefix.concat(conntrNo);

        // 客户经理今日客户数
        String conntrCntCacheName = conntrCntPrefix.concat(dayId);

        // 公共揽存缓存客户号关系
        String pbRelationshipCacheName = pbRelationshipPrefix.concat(opacInstnNo);

        // 公共揽存今日客户数缓存
        String pbCntDayIdCacheName = pbCntPrefix.concat(dayId);

        // 机构今日客户数缓存
//        String orgCntDayIdCacheName = orgCntPrefix.concat(dayId);

        String baseName = dayId.concat(":").concat(realOrgCode);

        // 机构今日开户客户号缓存
        String orgOpacCustCacheName = orgOpacCustPrefix.concat(baseName);

        // 叶子机构今日销户客户号缓存
        String orgCacctCustCacheName = orgCacctCustPrefix.concat(baseName);

        // 账号列表： 默认无账号
        Set<String> allAcctNos = Sets.newHashSet();

        if (opacType.equals("01")) {
            // 开户
            // 不存在客户经理
            if (conntrNo.isEmpty()) {
                String acctsStr = pbRelationshipDb.hget(pbRelationshipCacheName, custNo);
                if (acctsStr == null) {
                    allAcctNos.add(acctNo);
                    pbRelationshipDb.hset(pbRelationshipCacheName, custNo, setToStr(allAcctNos), null);
                    // TODO: 2022/09/15 客户数 +1 ：数据放出
                    // 公共揽存增加客户数，获取今日公共揽存客户数
                    rtDb.hincrBy(pbCntDayIdCacheName, opacInstnNo, 1L, ttl);
                    // 机构增加客户数，获取机构客户数
//                    rtDb.hincrBy(orgCntDayIdCacheName, realOrgCode, 1L, ttl);
                    rtDb.sadd(orgOpacCustCacheName, custNo);

                } else {
                    allAcctNos = strToSet(acctsStr);
                    boolean contains = allAcctNos.contains(acctNo);
                    if (!contains) {
                        allAcctNos.add(acctNo);
                        pbRelationshipDb.hset(pbRelationshipCacheName, custNo, setToStr(allAcctNos), null);
                        // TODO: 2022/09/15 数据放出
                    } else {
                        // There's nothing to do
                    }
                }
            } else {
                // 存在客户经理
                String acctsStr = conntrRelationshipDb.hget(conntrRelationshipCacheName, custNo);
                if (acctsStr == null) {
                    allAcctNos.add(acctNo);
                    conntrRelationshipDb.hset(conntrRelationshipCacheName, custNo, setToStr(allAcctNos), null);
                    // TODO: 2022/09/15 客户数 +1 ：数据放出
                    // 客户经理客户数+1
                    rtDb.hincrBy(conntrCntCacheName, conntrNo, 1L, ttl);

                    // 机构客户数 +1
//                    rtDb.hincrBy(orgCntDayIdCacheName, realOrgCode, 1L, ttl);

                    // 机构今日客户号缓存+1
                    rtDb.sadd(orgOpacCustCacheName, custNo);

                } else {
                    // 存在老账号
                    allAcctNos = strToSet(acctsStr);
                    boolean contains = allAcctNos.contains(acctNo);
                    if (!contains) {
                        allAcctNos.add(acctNo);
                        conntrRelationshipDb.hset(conntrRelationshipCacheName, custNo, setToStr(allAcctNos), null);
                        // TODO: 2022/09/15 数据放出
                    } else {
//                    return null;
                    }
                }
            }
        } else {
            // 销户
            if (conntrNo.isEmpty()) {
                // 不存在客户经理
                // 公共揽存数据
                String acctsStr = pbRelationshipDb.hget(pbRelationshipCacheName, custNo);
                if (acctsStr == null) {
                    // TODO: 2022/09/15 缓存中客户数 -1  ：数据放出 有待考究
                    // 公共揽存客户数 -1
                    rtDb.hincrBy(pbCntDayIdCacheName, opacInstnNo, -1L, ttl);
                    // 机构客户数 -1
//                    rtDb.hincrBy(orgCntDayIdCacheName, realOrgCode, -1L, ttl);

                    // 机构今日销户客户号缓存
                    rtDb.sadd(orgCacctCustCacheName, custNo);
                } else {
                    allAcctNos = strToSet(acctsStr);
                    int size = allAcctNos.size();
                    boolean contains = allAcctNos.contains(acctNo);
                    if (size == 1) {
                        if (contains) {
                            pbRelationshipDb.hdel(pbRelationshipCacheName, custNo);
                            // TODO: 2022/09/15 缓存中客户数减一 ：数据放出
                            // 公共揽存客户数 -1
                            rtDb.hincrBy(pbCntDayIdCacheName, opacInstnNo, -1L, ttl);
                            // 机构客户数 -1
//                            rtDb.hincrBy(orgCntDayIdCacheName, realOrgCode, -1L, ttl);
                            // 机构今日销户客户号缓存
                            rtDb.sadd(orgCacctCustCacheName, custNo);
                        } else {
                            // TODO: 2022/09/15  缓存中客户数不变 ：数据放出 有待考究
//                        return null;
                        }
                    } else {
                        if (contains) {
                            allAcctNos.remove(acctNo);
                            pbRelationshipDb.hset(pbRelationshipCacheName, custNo, setToStr(allAcctNos), null);
                            // TODO: 2022/09/15 缓存中客户数不变 ：数据放出 暂无异议
                        } else {
                            // TODO: 2022/09/15  缓存中客户数不变 ：数据放出 有待考究
//                        return null;
                        }
                    }
                }
            } else {
                // 存在客户经理
                String acctsStr = conntrRelationshipDb.hget(conntrRelationshipCacheName, custNo);
                if (acctsStr == null) {
                    // TODO: 2022/09/15 缓存中客户数 -1  ：数据放出 有待考究
                    rtDb.hincrBy(conntrCntCacheName, conntrNo, -1L, ttl);

                    // 机构客户数 -1
//                    rtDb.hincrBy(orgCntDayIdCacheName, realOrgCode, -1L, ttl);

                    // 机构今日销户客户号缓存
                    rtDb.sadd(orgCacctCustCacheName, custNo);
                } else {
                    allAcctNos = strToSet(acctsStr);
                    int size = allAcctNos.size();
                    boolean contains = allAcctNos.contains(acctNo);
                    if (size == 1) {
                        if (contains) {
                            conntrRelationshipDb.hdel(conntrRelationshipCacheName, custNo);
                            // TODO: 2022/09/15 缓存中客户数减一 ：数据放出
                            rtDb.hincrBy(conntrCntCacheName, conntrNo, -1L, ttl);
                            // 机构客户数 -1
//                            rtDb.hincrBy(orgCntDayIdCacheName, realOrgCode, -1L, ttl);
                            // 机构今日销户客户号缓存
                            rtDb.sadd(orgCacctCustCacheName, custNo);
                        } else {
                            // TODO: 2022/09/15  缓存中客户数不变 ：数据放出 有待考究
//                        return null;
                        }
                    } else {
                        if (contains) {
                            allAcctNos.remove(acctNo);
                            conntrRelationshipDb.hset(conntrRelationshipCacheName, custNo, setToStr(allAcctNos), null);   // 设置一天的过期时间
                            // TODO: 2022/09/15 缓存中客户数不变 ：数据放出 暂无异议
                        } else {
                            // TODO: 2022/09/15  缓存中客户数不变 ：数据放出 有待考究
//                        return null;
                        }
                    }
                }
            }
        }

        in.setValue(param.getCbOrgPdCnt(), getCustCnt(pbCntDayIdCacheName, opacInstnNo));
        in.setValue(param.getCbConntrCnt(), getCustCnt(conntrCntCacheName, conntrNo));
//        in.setValue(param.getCbOrgCnt(), getCustCnt(orgCntDayIdCacheName, realOrgCode));

        Long currentCustCnt = rtDb.scard(orgOpacCustCacheName) - rtDb.scard(orgCacctCustCacheName);

        in.setValue(param.getCbOrgCnt(), currentCustCnt);


        return in;
    }


    @Override
    public void close() throws Exception {
        super.close();
        pbRelationshipDb.close();
        conntrRelationshipDb.close();
        rtDb.close();
    }


    private Long getCustCnt(String cacheName, String hashKey) {
        long res = 0;
        String cntStr = rtDb.hget(cacheName, hashKey);
        if (cntStr != null) {
            res = Long.parseLong(cntStr);
        }
        return res;
    }


    @SneakyThrows
    private Set<String> strToSet(String jsonStr) {
        if (jsonStr == null || jsonStr.isEmpty()) {
            return Sets.newHashSet();
        }
        return om.readValue(jsonStr, new TypeReference<Set<String>>() {
        });
    }

    @SneakyThrows
    private String setToStr(Set<String> set) {
        return om.writeValueAsString(set);
    }

    private SimpleRedisContainer buildContainer(Integer db) {
        FlinkJedisSentinelConfig pbRelationshipDbConf =
                new FlinkJedisSentinelConfig.Builder()
                        .setMasterName(param.getMasterName())
                        .setSentinels(param.getSentinels())
                        .setPassword(param.getPassword())
                        .setDatabase(db)
                        .build();

        return build(pbRelationshipDbConf);
    }
}
